package sparkRDD

import org.apache.spark.{SparkConf, SparkContext}


object StudySpark {

  def main(args: Array[String]): Unit = {


    // spark  操作对象 是 rdd 弹性分布式数据集 （类似于 array ）
    // 逻辑：
    // 1)需要 sparkconf  配置  本地，集群yarn-client ，yarn-cluster 模式
    //            任务，名字
    //  2)  配置，sparkContext   sc  是spark 运行代码的入口
    //  maven 导入 spark-core 的依赖

    // rdd   Resilient Distributed Dataset  弹性分布式数据集
    // 弹性  转换类型，动作  （转换类型 不会运行出结果 map；；动作类型算子  foreach ，count，sum  ）
    // 分布式   分布在不同节点node 进行计算
    // 数据集   操作的不在是单一的某一个变量，操作对象是 一个集合




    val conf = new SparkConf().setMaster("local[*]").setAppName("helloSpark")
    val sc = new SparkContext(conf)    // sparkContext 的上下文  ，才能够运行spark 的任务

    val stringData = Array("hello spark","hello scala","hello java") // scala数组

    val strRDD = sc.parallelize(stringData) // 将数组 转换为 rdd
    // strRDD  ("hello spark"),("hello scala"),("hello java")
    // flatMap  x=>x.split(" ")  --> "hello spark", "hello scala"  --->     (hello spark)  (hello scala)
    //  flatMap    (hello spark)  (hello scala)  --->  hello spark  hello scala


    val strObj = strRDD.flatMap( x=>x.split(" ") )   // hello spark hello scala
    // flatMap  将rdd当中每个元素 ，切分 压平

    val mapObj = strObj.map( x=>(x,1))   // (hello,1)  (spark, 1) (hello ,1 )...
    // map   x=>(x,1)   hello spark  hello scala ---> (hello,1 )  (spark,1) (hello ,1) ( scala,1)

    val keyObj = mapObj.reduceByKey((x,y)=>(x+y)) // (hello,3) (spark ,1)...
    // reduceByKey  (x,y)=>(x+y)  (hello,1 )  (spark,1) (hello ,1) ( scala,1) ---> hello  1+1+1 -->(hello,3)
    //   keyObj.foreach(    x=> {println(x)  }   )

    //   keyObj.foreach(   ( x:(String,Int)) => {println(x)  }  )
    keyObj.foreach( println(_) )





    // (hello,3)  (scala,1)  (spark,1)  --->遍历 打印

    //1 def  printFunction ( x:(String,Int) ) = {     println(x)  }
    //2  ( x:(String,Int) ) => {     println(x)  }
    //3  x => {     println(x)  }
    //4  println(_)    _ 下划线 是通配符



    //  ( x:(String,Int)) => {println(x)  }


//最简单
    strRDD.flatMap(_.split("")).map((_,1)).reduceByKey((_+_)).foreach(println(_))


strRDD.filter(x=>x.contains("hello")).distinct().map((_,1)).reduceByKey(_+_).foreach(
 //println(_)函数
  
  x=>{
   var b = 0
   b = b + x._2
   println(b)
 }
)
















  }

}

